package cn.sciento.transfer.infra.rocketmq;

import cn.sciento.core.exception.CommonException;import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import cn.sciento.transfer.infra.rocketmq.config.MqConfigProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class RocketmqProducersGroupManager {
  private final Logger logger = LoggerFactory.getLogger(getClass());
  
  private final Map<String, DefaultMQProducer> producersGroup = new HashMap<>();
  
  private final Map<String, Boolean> started = new HashMap<>();
  
  @Autowired
  private MqConfigProperties mqConfigProperties;
  
  public synchronized DefaultMQProducer getOrCreateProducer(String group) {
    if (this.producersGroup.containsKey(group)) {
      return this.producersGroup.get(group);
    }
    DefaultMQProducer producer = new DefaultMQProducer(group);
    producer.setNamesrvAddr(this.mqConfigProperties.getNameSrv());
    this.producersGroup.put(group, producer);
    this.started.put(group, Boolean.FALSE);
    this.logger.info("RocketMQ CREATED for PCS group {} created", group);
    return producer;
  }
  
  public synchronized void startProducer() throws MQClientException {
    for (String group : getProducerGroups()) {
      start(group);
    }
  }
  
  public synchronized void startProducer(String producer) {
    start(producer);
  }
  
  public synchronized void stopProducer(String group) {
    stop(group);
  }
  
  private void stop(String group) {
    if (this.producersGroup.get(group) != null) {
      this.producersGroup.get(group).shutdown();
      this.started.put(group, Boolean.FALSE);
    } 
  }
  
  private synchronized void start(String group) {
    if (this.started.get(group) == null || this.started.get(group)) {
      return;
    }
    try {
      this.producersGroup.get(group).start();
      this.started.put(group, Boolean.TRUE);
    } catch (MQClientException e) {
      this.logger.error("RocketMQ Produce hasn't been started. Caused by " + e
          .getErrorMessage(), (Throwable)e);
      throw new CommonException(e);
    } 
  }
  
  public synchronized Set<String> getProducerGroups() {
    return this.producersGroup.keySet();
  }
}
